package io.github.quickmsg.core.protocol;

import io.github.quickmsg.common.acl.AclAction;
import io.github.quickmsg.common.acl.AclManager;
import io.github.quickmsg.common.channel.MqttChannel;
import io.github.quickmsg.common.context.ContextHolder;
import io.github.quickmsg.common.context.ReceiveContext;
import io.github.quickmsg.common.integrate.SubscribeTopic;
import io.github.quickmsg.common.integrate.msg.IntegrateMessages;
import io.github.quickmsg.common.integrate.topic.IntegrateTopics;
import io.github.quickmsg.common.log.LogEvent;
import io.github.quickmsg.common.log.LogManager;
import io.github.quickmsg.common.log.LogStatus;
import io.github.quickmsg.common.message.RetainMessage;
import io.github.quickmsg.common.message.mqtt.SubscribeMessage;
import io.github.quickmsg.common.metric.CounterType;
import io.github.quickmsg.common.protocol.Protocol;
import io.github.quickmsg.common.utils.JacksonUtil;
import io.github.quickmsg.common.utils.MqttMessageUtils;
import io.netty.handler.codec.mqtt.MqttQoS;
import io.netty.handler.codec.mqtt.MqttReasonCodes;
import reactor.util.context.ContextView;

import java.util.List;
import java.util.Set;

/**
 * @author luxurong
 */
public class SubscribeProtocol implements Protocol<SubscribeMessage> {


    @Override
    public void parseProtocol(SubscribeMessage message, MqttChannel mqttChannel, ContextView contextView) {
        ContextHolder.getReceiveContext().getMetricManager().getMetricRegistry().getMetricCounter(CounterType.SUBSCRIBE_EVENT).increment();

        ReceiveContext<?> receiveContext = contextView.get(ReceiveContext.class);
        LogManager logManager = receiveContext.getLogManager();
        IntegrateTopics<SubscribeTopic> topics = receiveContext.getIntegrate().getTopics();
        AclManager aclManager = receiveContext.getAclManager();
        IntegrateMessages messages = receiveContext.getIntegrate().getMessages();

        List<SubscribeTopic> subscribeTopics = message.getSubscribeTopics();
        int subscribeTopicsSize = subscribeTopics.size();
        MqttReasonCodes.SubAck[] subAcks = new MqttReasonCodes.SubAck[subscribeTopicsSize];
        for (int i = 0; i < subscribeTopicsSize; i++) {
            SubscribeTopic subscribeTopic = subscribeTopics.get(i);
            boolean isAcl = aclManager.check(mqttChannel, subscribeTopic.getTopicFilter(), AclAction.SUBSCRIBE);
            if (isAcl) {
                subAcks[i] = MqttReasonCodes.SubAck.valueOf((byte) subscribeTopic.getQoS().value());
                this.loadRetainMessage(messages, subscribeTopic);
            } else {
                subAcks[i] = MqttReasonCodes.SubAck.NOT_AUTHORIZED;
            }
        }

        topics.registryTopic(mqttChannel, subscribeTopics);
        logManager.printInfo(mqttChannel, LogEvent.SUBSCRIBE, LogStatus.SUCCESS, JacksonUtil.bean2Json(message));
        mqttChannel.write(MqttMessageUtils.buildSubAck(message.getMessageId(), subAcks));
    }

    @Override
    public Class<SubscribeMessage> getClassType() {
        return SubscribeMessage.class;
    }


    /**
     * 加载并且发送保留消息
     *
     * @param messages
     * @param topic
     */
    private void loadRetainMessage(IntegrateMessages messages, SubscribeTopic topic) {
        Set<RetainMessage> retainMessages = messages.getRetainMessage(topic.getTopicFilter());

        retainMessages.forEach(retainMessage -> {
            if (topic.getMqttChannel() != null) {
                topic.getMqttChannel().sendPublish(topic.minQos(MqttQoS.valueOf(retainMessage.getQos())),
                        retainMessage.toPublishMessage());
            }
        });
    }


}
